package com.xinqing.bigdata.flink.datastream.function;

import com.xinqing.bigdata.flink.datastream.model.UserCostInfo;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @Author:CHQ
 * @Date:2021/5/24 15:16
 * @Description
 */
public class AvgAggreFunction implements AggregateFunction<UserCostInfo, Tuple2<Double, Long>, Double> {

    Logger logger = LoggerFactory.getLogger(AvgAggreFunction.class);

    @Override
    public Tuple2<Double, Long> createAccumulator() {
        //新窗口初始化计数器
        return new Tuple2<>(0d, 0L);
    }

    //开窗中流入元素后调用此方法，无论是否关窗
    @Override
    public Tuple2<Double, Long> add(UserCostInfo value, Tuple2<Double, Long> accumulator) {
        Tuple2<Double, Long> t1 = new Tuple2<>(accumulator.f0 + value.getMoney(), accumulator.f1 + 1L);
        logger.info("add方法返回元组为->{}", t1.toString());
        return t1;
    }

    //根据自定义逻辑获取当前窗口结果
    @Override
    public Double getResult(Tuple2<Double, Long> accumulator) {
        double v = accumulator.f0 / accumulator.f1;
        logger.info("getResult方法返回元组为->{}", v);
        return v;
    }

    //此方法不能merge多个窗口，原因暂时不详
    @Override
    public Tuple2<Double, Long> merge(Tuple2<Double, Long> a, Tuple2<Double, Long> b) {
        logger.info("merge方法中的a元组为->{}", a.toString());
        logger.info("merge方法中的b元组为->{}", b.toString());
        Tuple2<Double, Long> t2 = new Tuple2<>((a.f0 + b.f0), a.f1 + b.f1);
        logger.info("merge方法中返回元组为->{}", t2.toString());
        return t2;
    }
}
